package com.flink.top_N;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import com.flink.window.WindowFunctionDemo;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.*;

/**
 * 描述:
 * 通过窗口函数统计用户访问量并排序
 *
 * @author yanzhengwu
 * @create 2022-08-07 23:52
 */
public class TopNProcessAllWindowFunction {

    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);
        //设置水位线生成的间隔 这里给的是100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
//        env.getConfig().setAutoWatermarkInterval(100);


        //TODO 无序流的watermark生成策略
        DataStream<User> stream = env.addSource(new UserSource(100))       //生成水位线和时间戳的策略对象
                .assignTimestampsAndWatermarks(
                        //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                        WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                //返回策略对象的具体实现
                                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                    /**
                                     * 此方法为指定以事件时间的具体时间戳字段
                                     *
                                     * @param element
                                     * @param recordTimestamp
                                     * @return 返回的则是一个毫秒数的时间戳
                                     */
                                    @Override
                                    public long extractTimestamp(User element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                }));

        stream.map(User::getName)
                //TODO 当前只是作为测试使用，正式生产环境 尽量不要使用windowAll 会造成严重数据倾斜，效率极低
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                //TODO 计算窗口由aggregate窗口聚合函数进行处理，完成后调用allWindowFunction进行最后的 输出规则
                .aggregate(new MyAggregateFunction(), new MyAllWindowFunction())
                .print();


        env.execute();


    }

    //TODO 自定义增量聚合函数,编写统计规则，此处使用hashMap则为统一收集进行处理，会造成数据倾斜
    public static class MyAggregateFunction implements AggregateFunction<String, HashMap<String, Long>, List<Tuple2<String, Long>>> {


        @Override
        public HashMap<String, Long> createAccumulator() {
            return new HashMap<>();
        }

        @Override
        public HashMap<String, Long> add(String value, HashMap<String, Long> accumulator) {

            if (accumulator.containsKey(value)) {
                Long count = accumulator.get(value);
                accumulator.put(value, count + 1);
            } else {
                accumulator.put(value, 0L);
            }
            return accumulator;
        }

        @Override
        public List<Tuple2<String, Long>> getResult(HashMap<String, Long> accumulator) {
            List<Tuple2<String, Long>> list = new ArrayList<>();
            for (Map.Entry<String, Long> entry : accumulator.entrySet()) {
                list.add(Tuple2.of(entry.getKey(), entry.getValue()));
            }

            list.sort(new Comparator<Tuple2<String, Long>>() {
                @Override
                public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                    //降序排列==后者 - 前者 （此处数量过大会造成数据不准确，实际生产环境下要考量）
                    return o2.f1.intValue() - o1.f1.intValue();
                }
            });


            return list;
        }

        @Override
        public HashMap<String, Long> merge(HashMap<String, Long> a, HashMap<String, Long> b) {
            return null;
        }
    }
    //TODO 全窗口函数的信息输出规则
    public static class MyAllWindowFunction extends ProcessAllWindowFunction<List<Tuple2<String, Long>>, String, TimeWindow> {

        @Override
        public void process(Context context, Iterable<List<Tuple2<String, Long>>> elements, Collector<String> out) throws Exception {
            long start = context.window().getStart();
            long end = context.window().getEnd();


            StringBuilder sb = new StringBuilder();
            sb.append("----------------------\n")
                    .append("窗口开始时间：" + new Timestamp(start))
                    .append("窗口结束时间：" + new Timestamp(end) + "\n");
            List<Tuple2<String, Long>> list = elements.iterator().next();
            for (int i = 0; i < 2; i++) {
                if (i < list.size()) {
                    Tuple2<String, Long> t = list.get(i);
                    String str = "No." + (i + 1) + "\n"
                            + "user:" + t.f0 + "\n"
                            + "count:" + t.f1 + "\n";
                    sb.append(str);
                }


            }
            sb.append("----------------------\n");
            out.collect(sb.toString());


        }
    }
}
